package com.ibm.cps.newmessage;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.log4j.Logger;

import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.processors.ProcessorFactory;
import com.ibm.cps.processors.deletion.BFSSearch;
import com.ibm.cps.processors.deletion.PartialSearch;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;

public class MetadataFactory {

	private static Logger logger = Logger.getLogger(MetadataFactory.class);

	public enum MetadataType {
		DataSourceMetadata, BasicProcessorMetadata, OptimizedProcessorMetadata, JointProcessorMetadata
	}

	public static AbstractMetadata parseJsonMetadata(String object)
			throws CPSException {
		return parseJsonMetadata(JSONUtility.fromObjectString(object));
	}

	public static AbstractMetadata parseJsonMetadata(ObjectNode object)
			throws CPSException {
		ProcessorFactory factory = ProcessorFactory.getInstance();
		Map<String, MetadataType> type2Metadata = factory.getProcessorType();
		String type = object.get(MessageFields.TYPE).asText();
		MetadataType metadataType = type2Metadata.get(type);
		if (metadataType == null) {
			throw new CPSException(ErrorCode.INVALID_INPUT_ERROR,
					"unregistered meta data type " + type);
		}
		switch (metadataType) {
		case BasicProcessorMetadata:
			return new BasicMetadata(object);
		case OptimizedProcessorMetadata:
			return new OptimizedMetadata(object);
		case JointProcessorMetadata:
			return new JointMetadata(object);
		default:
			throw new CPSException(ErrorCode.INVALID_INPUT_ERROR,
					"Not valide input type " + type + ".");
		}
	}

	public static void updateProcessorMap(
			HashMap<String, Set<String>> processor2Successors,
			AbstractMetadata metadata) throws CPSException {
		if (metadata instanceof BasicMetadata) {
			updateBasicProcessorMap(processor2Successors, metadata);
		} else if (metadata instanceof OptimizedMetadata) {
			updateOptimizedProcessorMap(processor2Successors, metadata);
		} else if (metadata instanceof JointMetadata) {
			// TODO
		} else {
			throw new CPSException(ErrorCode.INVALID_INPUT_ERROR,
					"Not validate input.");
		}
	}

	private static void updateOptimizedProcessorMap(
			HashMap<String, Set<String>> processor2Successors,
			AbstractMetadata metadata) {

		logger.info("Processorid is " + metadata.getProcessorid()
				+ " parent id is "
				+ Arrays.toString(metadata.getParentProcessors().toArray()));
		if (metadata.getProcessorid() != null
				&& metadata.getParentProcessors() != null) {
			String optimizedParentId;
			for (String parent : metadata.getParentProcessors()) {
				optimizedParentId = parent + metadata.getInputFieldType();
				if (processor2Successors.get(optimizedParentId) == null) {
					processor2Successors.put(optimizedParentId,
							new HashSet<String>());
				}
				processor2Successors.get(optimizedParentId).add(
						metadata.getProcessorid());
				if (processor2Successors.get(parent) == null) {
					processor2Successors.put(parent, new HashSet<String>());
				}
				processor2Successors.get(parent).add(optimizedParentId);
			}
		}
	}

	private static void updateBasicProcessorMap(
			HashMap<String, Set<String>> processor2Successors,
			AbstractMetadata metadata) {
		BasicMetadata basicMetadata = (BasicMetadata) metadata;
		if (basicMetadata.processorid != null
				&& basicMetadata.parentProcessors != null) {
			for (String parent : metadata.getParentProcessors()) {
				if (processor2Successors.get(parent) == null) {
					processor2Successors.put(parent, new HashSet<String>());
				}
				processor2Successors.get(parent).add(basicMetadata.processorid);
			}
		}
	}

	/**
	 * delete all processors whose root is metadata
	 * @param parent2Successors
	 * @param metadata the processor which need to be deleted
	 * @return all affected processors in the format of <causeProcessorid, deletedProcessorid>
	 * @throws CPSException
	 */
	public static List<List<Object>> deleteProcessor(
			HashMap<String, Set<String>> parent2Successors,
			AbstractMetadata metadata) throws CPSException {
		String processorid = metadata.getProcessorid();
		if (metadata instanceof BasicMetadata) {
			List<String> affectedProcessors = getBasicSuccessors(processorid,
					parent2Successors);
			removeElements(parent2Successors, affectedProcessors);
			return getUpdateMessages(affectedProcessors, metadata);
		} else if (metadata instanceof OptimizedMetadata) {
			List<String> affectedProcessors = getBasicSuccessors(processorid,
					parent2Successors);
			List<String> retProcessors = getOptimizedSuccessors(processorid,
					parent2Successors);
			removeElements(parent2Successors, affectedProcessors);
			return getUpdateMessages(retProcessors, metadata);
		} else {
			return null;
		}
	}

	private static void removeElements(
			HashMap<String, Set<String>> parent2Successors,
			List<String> affectedProcessors) {
		if (parent2Successors == null) {
			return;
		}
		// remove all processors in subset
		for (String parent : parent2Successors.keySet()) {
			parent2Successors.get(parent).removeAll(affectedProcessors);
		}

		// remove all processors in key
		for (String parent : affectedProcessors) {
			parent2Successors.remove(parent);
		}
	}

	public static List<String> getBasicSuccessors(String processorid,
			HashMap<String, Set<String>> parent2Successors) {
		return BFSSearch.getBFSSuccessors(processorid, parent2Successors);
	}

	public static List<String> getOptimizedSuccessors(String processorid,
			HashMap<String, Set<String>> parent2Successors) {
		List<String> bfs = BFSSearch.getBFSSuccessors(processorid,
				parent2Successors);
		List<String> partial = PartialSearch.getAffectedProcessors(processorid,
				parent2Successors);
		bfs.addAll(partial);
		return bfs;
	}

	/**
	 * send affected processors to next bolt
	 * @param affectedProcessors
	 * @return
	 * @throws CPSException 
	 */
	private static List<List<Object>> getUpdateMessages(
			List<String> affectedProcessors, AbstractMetadata metadata)
			throws CPSException {
		List<List<Object>> retMsg = new ArrayList<List<Object>>();
		logger = Logger.getLogger(MetadataFactory.class);
		for (String successor : affectedProcessors) {
			logger.info("Affect processor " + successor);
			List<Object> newObjects = new ArrayList<Object>();
			newObjects.add(metadata.getProcessorid());
			newObjects.add(successor);
			retMsg.add(newObjects);
		}
		return retMsg;
	}

}
